use std::{
    collections::{HashMap, HashSet},
    fmt::{Display, Formatter, Result},
    fs::File,
};

use tokio::io::BufReader;

use super::Command;

#[derive(Debug, Clone, Default)]
pub struct Metadata {
    pub topics: HashMap<String, MqTopic>,   //key是topic
    pub groups: HashMap<String, MqGroup>,   //key是topic
    pub messages: HashMap<String, Message>, //key是topic
}

#[derive(Debug, Clone, Default)]
pub struct MqGroup {
    pub groups: HashSet<String>,
}

#[derive(Debug, Clone, Default)]
pub struct PartitionMessage {
    pub partition: u16,
    pub msg_files: Vec<String>,
}

#[derive(Debug, Clone, Default)]
pub struct Message {
    pub topic: String,
    pub partition_map: HashMap<u16, PartitionMessage>, //key是partition
}

#[derive(Debug, Clone, Default)]
pub struct MqTopic {
    pub topic: String,
    pub partitions: u16,
    pub replication: u16, //先不实现
}

impl Metadata {
    pub fn init_topic(&mut self, topic: MqTopic) {
        let message_topic = topic.topic.to_owned();
        self.topics.insert(topic.topic.clone(), topic);
        self.messages.insert(message_topic, Message::default());
    }

    pub fn get_all_topic(&self) -> Option<Vec<String>> {
        Some(self.topics.keys().map(|k| (*k).clone()).collect())
    }

    pub fn remove_topic(&mut self, topic: &str) {
        self.topics.remove(topic);
    }

    pub fn get_topic(&mut self, topic: &str) -> Option<&MqTopic> {
        self.topics.get(topic)
    }

    pub fn change_partitions(&mut self, topic: &str, partition: u16) -> Option<&MqTopic> {
        let mq_topic = self.topics.get_mut(topic);
        if let Some(mq) = mq_topic {
            mq.partitions = partition;
        }
        self.topics.get(topic)
    }

    pub fn add_group(&mut self, topic: &str, group: String) {
        let groups_option = self.groups.get_mut(topic);
        match groups_option {
            Some(my_group) => {
                my_group.groups.insert(topic.to_owned());
            }
            None => {
                let mut my_group = MqGroup::default();
                my_group.groups.insert(group.to_owned());
                self.groups.insert(topic.to_owned(), my_group);
            }
        }
    }

    pub fn remove_group(&mut self, group: &str) {
        self.groups.remove(group);
    }
}

impl MqTopic {
    pub fn new(command: Command) -> Self {
        MqTopic {
            topic: command.topic.unwrap().to_owned(),
            partitions: command.partitions.unwrap(),
            replication: command.replication_factor.unwrap(),
        }
    }
}

impl Display for MqTopic {
    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
        writeln!(f, "{{")?;
        writeln!(f, "\t topic: {}, ", self.topic)?;
        writeln!(f, "\t partition: {},", self.partitions)?;
        writeln!(f, "\t replication-factor: {}", self.replication)?;
        write!(f, "}}")
    }
}
